package org.zhadoop.streaming.util.datatype;

import java.sql.Date;
import java.sql.Time;
import java.text.SimpleDateFormat;
import java.util.TimeZone;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zhadoop.streaming.config.StreamingConfig;
import org.zhadoop.streaming.exception.StreamingException;

import com.google.common.base.Strings;

public class TimeParser implements DataTypeParser {

	/**
	 * 
	 */
	private static final long serialVersionUID = -4152678452115758152L;
	
	private static final Logger LOG = LoggerFactory.getLogger(TimeParser.class);
	
	private final SimpleDateFormat timeFormatter;
	
	private final TimeZone timeZone;

	public TimeParser(StreamingConfig config) throws StreamingException{
		timeZone = TimeZoneUtils.parseTimeZone(TimeConstants.ZERO_TIMEZONE);
		timeFormatter = new SimpleDateFormat(TimeConstants.TIME_FORMAT);
		timeFormatter.setTimeZone(timeZone);
		timeFormatter.setLenient(false);
	}
	
	@Override
	public Object createValue(String value) throws StreamingException {
		if(Strings.isNullOrEmpty(value)){
			return null;
		}
		
		if(value.length() > TimeConstants.TIME_FORMAT.length()){
			LOG.warn("Failed to create {} instance.",Date.class.getName());
			throw new StreamingException("Failed to create " + Date.class.getName() + " instance." );
		}
		
		try {
			java.util.Date uDate = timeFormatter.parse(value);
			return new java.sql.Time(uDate.getTime());
		} catch (Exception e) {
			LOG.warn("Failed to create {} instance.",Date.class.getName());
			throw new StreamingException("Failed to create " + Date.class.getName() + " instance." );
		}
	}

	@Override
	public String toStringValue(Object value) throws StreamingException {
		if(null == value){
			return null;
		}
		return timeFormatter.format(new Date(((Time)value).getTime()));
	}

	public TimeZone getTimeZone() {
		return timeZone;
	}

}
